// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#pragma once

#include <nebula/types/fundamental.h>

namespace nebula {
    /// \class Schema
    /// \brief sequence of nebula::Field objects describing the columns of a record
    /// batch or table data structure
    class TURBO_EXPORT Schema : public detail::FingerPrintAble,
                                public turbo::EqualityComparable<Schema>,
                                public turbo::ToStringOstreamable<Schema> {
    public:
        explicit Schema(FieldVector fields, Endianness endianness,
                        std::shared_ptr<const KeyValueMetadata> metadata = nullptr);

        explicit Schema(FieldVector fields,
                        std::shared_ptr<const KeyValueMetadata> metadata = nullptr);

        Schema(const Schema &);

        ~Schema() override;

        /// Returns true if all of the schema fields are equal
        bool equals(const Schema &other, bool check_metadata = false) const;

        bool equals(const std::shared_ptr<Schema> &other, bool check_metadata = false) const;

        /// \brief Set endianness in the schema
        ///
        /// \return new Schema
        std::shared_ptr<Schema> with_endianness(Endianness endianness) const;

        /// \brief Return endianness in the schema
        Endianness endianness() const;

        /// \brief Indicate if endianness is equal to platform-native endianness
        bool is_native_endian() const;

        /// \brief Return the number of fields (columns) in the schema
        int num_fields() const;

        /// Return the ith schema element. Does not boundscheck
        const std::shared_ptr<Field> &field(int i) const;

        const FieldVector &fields() const;

        std::vector<std::string> field_names() const;

        /// Returns null if name not found
        std::shared_ptr<Field> get_field_by_name(const std::string &name) const;

        /// \brief Return the indices of all fields having this name in sorted order
        FieldVector get_all_fields_by_name(const std::string &name) const;

        /// Returns -1 if name not found
        int get_field_index(const std::string &name) const;

        /// Return the indices of all fields having this name
        std::vector<int> get_all_field_indices(const std::string &name) const;

        /// Indicate if field named `name` can be found unambiguously in the schema.
        turbo::Status can_reference_field_by_name(const std::string &name) const;

        /// Indicate if fields named `names` can be found unambiguously in the schema.
        turbo::Status can_reference_fields_by_names(const std::vector<std::string> &names) const;

        /// \brief The custom key-value metadata, if any
        ///
        /// \return metadata may be null
        const std::shared_ptr<const KeyValueMetadata> &metadata() const;

        /// \brief Render a string representation of the schema suitable for debugging
        /// \param[in] show_metadata when true, if KeyValueMetadata is non-empty,
        /// print keys and values in the output
        std::string to_string(bool show_metadata = false) const;

        turbo::Result<std::shared_ptr<Schema> > add_field(int i,
                                                         const std::shared_ptr<Field> &field) const;

        turbo::Result<std::shared_ptr<Schema> > remove_field(int i) const;

        turbo::Result<std::shared_ptr<Schema> > set_field(int i,
                                                         const std::shared_ptr<Field> &field) const;

        /// \brief Replace field names with new names
        ///
        /// \param[in] names new names
        /// \return new Schema
        turbo::Result<std::shared_ptr<Schema> > with_names(const std::vector<std::string> &names) const;

        /// \brief Replace key-value metadata with new metadata
        ///
        /// \param[in] metadata new KeyValueMetadata
        /// \return new Schema
        std::shared_ptr<Schema> with_metadata(
            const std::shared_ptr<const KeyValueMetadata> &metadata) const;

        /// \brief Return copy of Schema without the KeyValueMetadata
        std::shared_ptr<Schema> remove_metadata() const;

        /// \brief Indicate that the Schema has non-empty KevValueMetadata
        bool has_metadata() const;

        /// \brief Indicate that the Schema has distinct field names.
        bool has_distinct_field_names() const;

    protected:
        std::string compute_fingerprint() const override;

        std::string compute_metadata_fingerprint() const override;

    private:
        class Impl;

        std::unique_ptr<Impl> impl_;
    };

    TURBO_EXPORT void print_to(const Schema &s, std::ostream *os);

    /// \brief Convenience class to incrementally construct/merge schemas.
    ///
    /// This class amortizes the cost of validating field name conflicts by
    /// maintaining the mapping. The caller also controls the conflict resolution
    /// scheme.
    class TURBO_EXPORT SchemaBuilder {
    public:
        // Indicate how field conflict(s) should be resolved when building a schema. A
        // conflict arise when a field is added to the builder and one or more field(s)
        // with the same name already exists.
        enum ConflictPolicy {
            // Ignore the conflict and append the field. This is the default behavior of the
            // Schema constructor and the `nebula::schema` factory function.
            CONFLICT_APPEND = 0,
            // Keep the existing field and ignore the newer one.
            CONFLICT_IGNORE,
            // Replace the existing field with the newer one.
            CONFLICT_REPLACE,
            // Merge the fields. The merging behavior can be controlled by `Field::MergeOptions`
            // specified at construction time. Also see documentation of `Field::merge_with`.
            CONFLICT_MERGE,
            // Refuse the new field and error out.
            CONFLICT_ERROR
        };

        /// \brief Construct an empty SchemaBuilder
        /// `field_merge_options` is only effective when `conflict_policy` == `CONFLICT_MERGE`.
        SchemaBuilder(
            ConflictPolicy conflict_policy = CONFLICT_APPEND,
            Field::MergeOptions field_merge_options = Field::MergeOptions::defaults());

        /// \brief Construct a SchemaBuilder from a list of fields
        /// `field_merge_options` is only effective when `conflict_policy` == `CONFLICT_MERGE`.
        SchemaBuilder(
            FieldVector fields, ConflictPolicy conflict_policy = CONFLICT_APPEND,
            Field::MergeOptions field_merge_options = Field::MergeOptions::defaults());

        /// \brief Construct a SchemaBuilder from a schema, preserving the metadata
        /// `field_merge_options` is only effective when `conflict_policy` == `CONFLICT_MERGE`.
        SchemaBuilder(
            const std::shared_ptr<Schema> &schema,
            ConflictPolicy conflict_policy = CONFLICT_APPEND,
            Field::MergeOptions field_merge_options = Field::MergeOptions::defaults());

        /// \brief Return the conflict resolution method.
        ConflictPolicy policy() const;

        /// \brief Set the conflict resolution method.
        void set_policy(ConflictPolicy resolution);

        /// \brief Add a field to the constructed schema.
        ///
        /// \param[in] field to add to the constructed Schema.
        /// \return A failure if encountered.
        turbo::Status add_field(const std::shared_ptr<Field> &field);

        /// \brief Add multiple fields to the constructed schema.
        ///
        /// \param[in] fields to add to the constructed Schema.
        /// \return The first failure encountered, if any.
        turbo::Status add_fields(const FieldVector &fields);

        /// \brief Add fields of a Schema to the constructed Schema.
        ///
        /// \param[in] schema to take fields to add to the constructed Schema.
        /// \return The first failure encountered, if any.
        turbo::Status add_schema(const std::shared_ptr<Schema> &schema);

        /// \brief Add fields of multiple Schemas to the constructed Schema.
        ///
        /// \param[in] schemas to take fields to add to the constructed Schema.
        /// \return The first failure encountered, if any.
        turbo::Status add_schemas(const std::vector<std::shared_ptr<Schema> > &schemas);

        turbo::Status add_metadata(const KeyValueMetadata &metadata);

        /// \brief Return the constructed Schema.
        ///
        /// The builder internal state is not affected by invoking this method, i.e.
        /// a single builder can yield multiple incrementally constructed schemas.
        ///
        /// \return the constructed schema.
        turbo::Result<std::shared_ptr<Schema> > finish() const;

        /// \brief Merge schemas in a unified schema according to policy.
        static turbo::Result<std::shared_ptr<Schema> > Merge(
            const std::vector<std::shared_ptr<Schema> > &schemas,
            ConflictPolicy policy = CONFLICT_MERGE);

        /// \brief Indicate if schemas are compatible to merge according to policy.
        static turbo::Status are_compatible(const std::vector<std::shared_ptr<Schema> > &schemas,
                                           ConflictPolicy policy = CONFLICT_MERGE);

        /// \brief reset internal state with an empty schema (and metadata).
        void reset();

        ~SchemaBuilder();

    private:
        class Impl;

        std::unique_ptr<Impl> impl_;
    };

    /// \brief Unifies schemas by merging fields by name.
    ///
    /// The behavior of field merging can be controlled via `Field::MergeOptions`.
    ///
    /// The resulting schema will contain the union of fields from all schemas.
    /// Fields with the same name will be merged. See `Field::MergeOptions`.
    /// - They are expected to be mergeable under provided `field_merge_options`.
    /// - The unified field will inherit the metadata from the schema where
    ///   that field is first defined.
    /// - The first N fields in the schema will be ordered the same as the
    ///   N fields in the first schema.
    /// The resulting schema will inherit its metadata from the first input schema.
    /// Returns an error if:
    /// - Any input schema contains fields with duplicate names.
    /// - Fields of the same name are not mergeable.
    TURBO_EXPORT
    turbo::Result<std::shared_ptr<Schema> > unify_schemas(
        const std::vector<std::shared_ptr<Schema> > &schemas,
        Field::MergeOptions field_merge_options = Field::MergeOptions::defaults());
} // namespace nebula
